package com.burges.net.dataStream.windows.windowsFunction

import java.lang

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/1/27 
  * 描述     ProcessFunction的基本使用
  */
object ProcessWindowFunctionDemo {

	def main(args: Array[String]): Unit = {
		val parameterTool = ParameterTool.fromArgs(args)

		val env = StreamExecutionEnvironment.getExecutionEnvironment

		val inputStream = env.fromElements((1, 21L), (3, 1L), (5, 4L))

		val staticStream = inputStream.keyBy(_._1).timeWindow(Time.seconds(10))
//        		.process(new StaticProcessFunction)
	}

	//定义StaticProcessFunction，根据窗口中的数据统计指标
	class StaticProcessFunction extends ProcessWindowFunction[(String,Int,Int), (String, Long, Long, Long, Long, Long), String, TimeWindow]{
		override def process(key: String,
		                     ctx: ProcessWindowFunction[(String, Int, Int), (String, Long, Long, Long, Long, Long), String, TimeWindow]#Context,
		                     vals: lang.Iterable[(String, Int, Int)],
		                     out: Collector[(String, Long, Long, Long, Long, Long)]): Unit = {
			//定义求和，最大值，最小值，平均值，窗口时间逻辑
//			val sum: Long = vals.map(_._2).sum
//			val min: Long = vals.map(_._2).min
//			val max: Long = vals.map(_._2).max
//			var avg: Long = sum / vals.size
			val windowEnd = ctx.window().getEnd
			//通过out.collect返回计算结果
//			out.collect((key, min, max,sum,avg, windowEnd))
		}
	}

}
